package org.amos.rocketmq.core;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.util.List;
import java.util.UUID;

@Slf4j
public class RocketMqSender {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送异步消息
     *
     * @param topic   消息Topic
     * @param msg 消息实体
     */
    public void send(String topic, String msg) {
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msg).build(), defaultSendCallback());
    }
    
    /**
     * 发送异步消息
     *
     * @param topic   消息Topic
     * @param msg 消息实体
     */
    public void send(String topic, Message<?> msg) {
        rocketMQTemplate.asyncSend(topic, msg, defaultSendCallback());
    }

    /**
     * 发送异步消息
     *
     * @param topic        消息Topic
     * @param msg      消息实体
     * @param sendCallback 回调函数
     */
    public void send(String topic, Message<?> msg, SendCallback sendCallback) {
        rocketMQTemplate.asyncSend(topic, msg, sendCallback);
    }

    /**
     * 发送异步消息
     *
     * @param topic        消息Topic
     * @param msg      消息实体
     * @param sendCallback 回调函数
     * @param timeout      超时时间
     */
    public void send(String topic, Message<?> msg, SendCallback sendCallback, long timeout) {
        rocketMQTemplate.asyncSend(topic, msg, sendCallback, timeout);
    }

    /**
     * 发送异步消息
     *
     * @param topic        消息Topic
     * @param msg      消息实体
     * @param sendCallback 回调函数
     * @param timeout      超时时间
     * @param delayLevel   延迟消息的级别
     */
    public void send(String topic, Message<?> msg, SendCallback sendCallback, long timeout, int delayLevel) {
        rocketMQTemplate.asyncSend(topic, msg, sendCallback, timeout, delayLevel);
    }

    /**
     * 同步批量发送消息
     * @param topic
     * @param msg
     * @param timeout
     */
    public void sendBatch(String topic, List<Message<?>> msg, long timeout) {
        this.rocketMQTemplate.syncSend(topic, msg, timeout);
    }

    /**
     * 同步发送延时消息
     *
     * @param topic
     * @param msg
     * @param timeout
     * @param delayLevel 延时等级：现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，
     *                   从1s到2h分别对应着等级 1 到 18，消息消费失败会进入延时消息队列
     *                   "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
     */
    public void sendDelay(String topic, Message<?> msg, long timeout, int delayLevel) {
        this.rocketMQTemplate.syncSend(topic, msg, timeout, delayLevel);
    }

    /**
     * 异步发送延时消息
     * @param topic
     * @param msg
     * @param sendCallback
     * @param timeout
     * @param delayLevel
     */
    public void sendDelayAsync(String topic, Message<?> msg, SendCallback sendCallback, long timeout, int delayLevel) {
        this.rocketMQTemplate.asyncSend(topic, msg, sendCallback, timeout, delayLevel);
    }


    /**
     * 发送顺序消息
     *
     * @param msg
     * @param topic
     * @param hashKey
     */
    public void sendOrderly(String topic, Message<?> msg, String hashKey) {
        rocketMQTemplate.syncSendOrderly(topic, msg, hashKey);
    }

    /**
     * 发送顺序消息
     *
     * @param msg
     * @param topic
     * @param hashKey
     * @param timeout
     */
    public void sendOrderly(String topic, Message<?> msg, String hashKey, long timeout) {
        rocketMQTemplate.syncSendOrderly(topic, msg, hashKey, timeout);
    }

    /**
     *
     * @param topic
     * @param msg
     * @return
     */
    public TransactionSendResult sendTxMsg(String topic, Object msg) {
        String transactionId = UUID.randomUUID().toString();
        TransactionSendResult result = this.rocketMQTemplate.sendMessageInTransaction(topic, MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
                .build(), msg);
        log.info("[RocketMqSender][sendTxMsg],result = {}",result.toString());
        return result;
    }


    /**
     * 默认CallBack函数
     *
     * @return
     */
    private SendCallback defaultSendCallback() {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("[RocketMqSender],发送消息失败:{}", throwable.getMessage());
            }
        };
    }
}
